package com.lxy.mq.consumer;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.lxy.es.bean.Product;
import com.lxy.es.repository.ProductRepository;
import com.lxy.mq.bean.CanalBean;
import com.lxy.mq.bean.ProductSqlField;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author lixueyang
 * @since 2022/4/20 14:10
 */
@Slf4j
@Service
@RocketMQMessageListener(
        consumerGroup = "canal_sync_group",
        topic = "canal_sync_topic",
        consumeMode = ConsumeMode.ORDERLY,
        maxReconsumeTimes = 2
)
public class CanalSyncConsumer implements RocketMQListener<String> {

    @Resource
    ProductRepository productRepository;

    @Override
    public void onMessage(String message) {

        log.info("同步的信息{}", message);

        CanalBean canalBean = JSON.parseObject(message, CanalBean.class);
        log.info("同步的信息{}", JSON.toJSONString(canalBean));

        List<ProductSqlField> infos = canalBean.getData().stream().map(it -> JSON.parseObject(it, ProductSqlField.class)).collect(Collectors.toList());

        for (ProductSqlField info : infos) {


            Product product = new Product();
            BeanUtil.copyProperties(info, product);
            product.setPubStatus(info.getPub_status());
            product.setCreateTime(info.getCreate_time());
            productRepository.save(product);
        }
    }
}
